package com.cetc.sdp.kmga.cs.util;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @description: 接收RabbitMQ 通知消息
 * @author： DengQiang
 * @date: 2017/9/12 9:14
 */
public class RabbitMQConsumer extends Thread {

    private String host;
    private int port = 5672;
    private String userName;
    private String password;
    private String queue;
    private String topic;
    private String exchange;
    private String virtualHost = "/";
    private String routingKey = "";
    private boolean sslEnabled = false;
    private boolean autoAck = false;
    private boolean requeuing = false;

    private Connection connection;
    private Channel channel;

    private List<Address> addresses = new ArrayList<>();

    private RabbitMQProducer.XT xt = RabbitMQProducer.XT.DEFAULT;

    private List<MessageListener> msgListeners;

    public RabbitMQConsumer() {
        super();
        this.msgListeners = new ArrayList<>();
    }

    public RabbitMQConsumer setHost(String host) {
        this.host = host;
        return this;
    }

    public RabbitMQConsumer addAddress(String ip, int port) {
        addresses.add(new Address(ip, port));
        return this;
    }

    public RabbitMQConsumer setPort(int port) {
        this.port = port;
        return this;
    }

    public RabbitMQConsumer setUserName(String userName) {
        this.userName = userName;
        return this;
    }

    public RabbitMQConsumer setPassword(String password) {
        this.password = password;
        return this;
    }

    public RabbitMQConsumer setVirtualHost(String virtualHost) {
        this.virtualHost = virtualHost;
        return this;
    }

    public RabbitMQConsumer setQueue(String queue) {
        this.queue = queue;
        return this;
    }

    public RabbitMQConsumer setTopic(String topic) {
        this.topic = topic;
        return this;
    }

    public RabbitMQConsumer setExchange(String exchange) {
        this.exchange = exchange;
        return this;
    }

    public RabbitMQConsumer setSslEnabled(boolean sslEnabled) {
        this.sslEnabled = sslEnabled;
        return this;
    }

    public RabbitMQConsumer setAutoAck(boolean autoAck) {
        this.autoAck = autoAck;
        return this;
    }

    public RabbitMQConsumer setRequeuing(boolean requeuing) {
        this.requeuing = requeuing;
        return this;
    }

    public RabbitMQConsumer addMsgListener(MessageListener msgListener) {
        this.msgListeners.add(msgListener);
        return this;
    }

    public RabbitMQConsumer setXt(RabbitMQProducer.XT xt) {
        this.xt = xt;
        return this;
    }

    public RabbitMQConsumer setRoutingKey(String routingKey) {
        this.routingKey = routingKey;
        return this;
    }

    @Override
    public void run() {
        DefaultConsumer consumer;

        try {
            connection = createRabbitMQConnection();
            channel = connection.createChannel();

            switch (xt) {
                case DEFAULT:
                    break;
                case FANOUT:
                    //接收端也声明一个fanout交换机
                    channel.exchangeDeclare(exchange, "fanout", true, false, null);
                    //channel.exchangeDeclarePassive() 可以使用该函数使用一个已经建立的exchange
                    //声明一个临时队列，该队列会在使用完比后自动销毁
                    queue = channel.queueDeclare().getQueue();
                    //将队列绑定到交换机,参数3无意义此时
                    channel.queueBind(queue, exchange, routingKey);
                    break;
                case DIRECT:
                    channel.exchangeDeclare(exchange, "direct", true, false, null);
                    channel.queueBind(queue, exchange, routingKey);
                    break;
                case TOPIC:
                    channel.exchangeDeclare(exchange, "topic", true, false, null);
                    channel.queueBind(queue, exchange, routingKey);
                    break;
                case HEADERS:
                    channel.exchangeDeclare(exchange, "headers", true, false, null);
                    queue = channel.queueDeclare().getQueue();
                    Map<String, Object> headers = new HashMap<String, Object>() {{
                        put("x-match", "all");//all==匹配所有条件，any==匹配任意条件
                    }};
                    channel.queueBind(queue, exchange, routingKey, headers);
                    break;
                default: break;
            }

        } catch (TimeoutException | IOException e) {
            e.printStackTrace();
            return;
        }

        consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try {
                    String message = new String(body, "UTF-8");
                    msgListeners.forEach(messageListener -> messageListener.fireMessage(message));
                    channel.basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    rejectMessage(envelope.getDeliveryTag());
                }
            }
        };
        try {
            channel.basicConsume(queue, autoAck, "SparkStreamingConsumer", consumer);
        } catch (IOException e) {
            e.printStackTrace();
            this.close();
        }
    }

    private Connection createRabbitMQConnection() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername(userName);
        factory.setPassword(password);
        factory.setVirtualHost(virtualHost);
        if (sslEnabled) {
            try {
                factory.useSslProtocol();
            } catch (NoSuchAlgorithmException | KeyManagementException e) {
                e.printStackTrace();
            }
        }
        if (host != null) {
            addresses.add(new Address(host, port));
        }
        return factory.newConnection(addresses);
    }

    private void rejectMessage(long deliveryTag) {
        try {
            channel.basicReject(deliveryTag, requeuing);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    private void close() {
        try {
            channel.close();
            connection.close();
        } catch (IOException | TimeoutException ex) {
            ex.printStackTrace();
        }
    }

}
